package dacp.etl.kafka.hdfs.process;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Joiner;
import com.google.common.base.Objects;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

import dacp.etl.kafka.hdfs.connect.utils.CommonSinkConnectorConfig;
import dacp.etl.kafka.hdfs.process.impl.DelProcessor;
import dacp.etl.kafka.hdfs.process.impl.TimeProcessor;
import io.confluent.common.config.ConfigException;

public class ProcessChain {
	private static final Logger log = LoggerFactory.getLogger(ProcessChain.class);
	private ProcessSubChain root;

	private Splitter spliter;
	private Joiner joiner;

	public String process(String ori) { 
		ProcessSubChain cur = root;
		if(cur == null) return ori;
		ArrayList<String> list = Lists.newArrayList();

		Iterator<String> iterator = spliter.split(ori).iterator();
		int i = 0;
		while (iterator.hasNext()) {
			String next = iterator.next();
			if (cur.getInx() == i) {
				list.add(cur.trans(next));
				cur = cur.next();
				if (cur == null) {
					break;
				}
			}
			i++;
		}
		return joiner.join(list);
	}

	public static ProcessChain build(CommonSinkConnectorConfig config) {
		ProcessChain chain = new ProcessChain();
		
		List<String> s1 = null;
		try {
			chain.spliter = Splitter.on(config.getString(CommonSinkConnectorConfig.PS_RECORD_VAL_SPLIT_CHAR));
			chain.joiner = Joiner.on(config.getString(CommonSinkConnectorConfig.PS_RECORD_VAL_JOIN_CHAR)); 
			s1 = config.getList(CommonSinkConnectorConfig.PS_SPLILT_COLS);
		} catch (ConfigException e) {
			return null;
		}
		
		if(s1.size() == 0){
			return null;
		}

		List<Integer> splitCols = Lists.newArrayList();
		for (String s : s1) {
			splitCols.add(Integer.parseInt(s));
		}
		Collections.sort(splitCols);

		boolean time = false;
		HashSet<Integer> h2 = Sets.newHashSet();
		try {
			List<String> s2 = config.getList(CommonSinkConnectorConfig.PS_TODATETIME_COLS);
			if (s2.size() > 0) {
				time = true;
				for (String s : s2) {
					h2.add(Integer.parseInt(s));
				}
			}
		} catch (ConfigException e) {
			/* is ok */}

		TimeProcessor tp = new TimeProcessor(config.getString(CommonSinkConnectorConfig.PS_TODATETIME_FMT));

		boolean del = false;
		HashSet<Integer> h3 = Sets.newHashSet();
		try {
			List<String> s3 = config.getList(CommonSinkConnectorConfig.PS_DELETE_CHAR_COLS);
			if (s3.size() > 0) {
				del = true;
				for (String s : s3) {
					h3.add(Integer.parseInt(s));
				}
			}
		} catch (ConfigException e) {
			/* is ok */}

		DelProcessor dp = new DelProcessor(config.getString(CommonSinkConnectorConfig.PS_DELETE_CHAR));

		log.info("build chain split : {}", splitCols.size());
		log.info("build chain trans : {}", h2.size());
		log.info("build chain del   : {}", h3.size());

		int i = 0;
		ProcessSubChain current = null;
		for (int colInx : splitCols) {
			ProcessSubChain subChain = new ProcessSubChain(colInx);

			if (time && h2.contains(colInx)) {
				subChain.addProcessor(tp);
			}
			if (del && h3.contains(colInx)) {
				subChain.addProcessor(dp);
			}
			if (i == 0) {
				chain.root = subChain;
			} else {
				current.setNext(subChain);
			}
			current = subChain;
			i++;
		}

		log.info("build.chain : {}", chain.toString());
		return chain;
	}

	@Override
	public String toString() {
		return Objects.toStringHelper(this).add("root", root).toString();
	}

}
